package com.herohead.backup;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.io.FileUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.Tuple;

import com.herohead.Misc;
import com.herohead.WorkerThreadV2;
import com.herohead.redis.GetRedisServer;
import com.herohead.redis.RedisKeys;
import com.herohead.redis.RedisVars;
import com.herohead.redis.SendLog;

public class CloneCatchUpV2 {


	static String programStart;
	static String programEnd;
	static String tempkey;



	static Long processID;
	static Transaction tx;

	public static void main(String[] args) {
		// TODO Auto-generated method stub

		programStart = (Misc.getTimestamp());

		// get RedisServer

		GetRedisServer grs = new GetRedisServer();
		//String redisHost = grs.getRedisHost();
		RedisVars.redisHost = "localhost";
		Jedis jd = new Jedis(RedisVars.redisHost);




		try {
			grs.checkForRedis(jd);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.exit(1);
		}

		Jedis jdpl = new Jedis(RedisVars.redisHost);





		//String fileInput = "src\\clone.txt";
		String fileInput = args[0];
		String targetClone = args[1];
		int threadCnt = Integer.parseInt(args[2]);
		String cloneName = args[3];


		//log.debug(fileInput);
		File file = new File(fileInput);

		// RedisKeys

		processID = jd.incr(RedisKeys.sProdCloneProcessID);
		final String processIDStr = String.valueOf(processID);

		String cloneKey = RedisKeys.slistProdClones + targetClone  + ":" + cloneName + ":" +  processIDStr;

		String fileTxt = null;


		SendLog sl = new SendLog();

		sl.setRedisHost(RedisVars.redisHost);

		//System.exit(0);
		ExecutorService executor = Executors.newFixedThreadPool(threadCnt);

		final List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();

		try {
			//System.out.println(file.length());
			if (file.length() == 0) {
				//sl.setLogandWrite("t:log:message", "info", "low", "cloneCatchUp", "cloneCatchUp", "cloneCatchUp-log", "No SSID found");
				System.out.println("No SSID found to clone");
				System.exit(0);
			}

			//sl.setLogandWrite("t:log:message", "info", "low", "cloneCatchUp", "cloneCatchUp", "cloneCatchUp-log", "Clone Started");

			jd.hset(RedisKeys.hashProdCloneRunningName, processIDStr,cloneName);
			tempkey = processIDStr + ":" + cloneName +":" + programStart + ":cloneCatchUp:" + targetClone;
			jd.rpush(RedisKeys.listProdCloneRunning, tempkey);

			tx = jdpl.multi();
			fileTxt = FileUtils.readFileToString(file, "UTF-8");
			String[] lines = fileTxt.split("\n");
			for (String line : lines) {
				if (line.contains("ssid")) {continue;}
				line = line.trim();
				line = line.replaceAll("\\s+", " ");
				String ssid = line.split(" ")[0];
				String size = line.split(" ")[1];
				tx.zadd(cloneKey, Double.parseDouble(size),ssid);
				tx.hset(RedisKeys.hashProdCloneSSID, ssid, size);
				//System.out.println(ssid + " " + size);
			}
			tx.exec();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		//System.exit(0);

		// anything over 50GB kick off single clone
		Set<Tuple> s = jd.zrevrangeByScoreWithScores(cloneKey, "+inf", "50000000000");
		tx = jdpl.multi();
		for (Tuple ss : s) {
			double d = ss.getScore();
			BigDecimal big = new BigDecimal(d);
			System.out.println(ss.getElement() + " " +  big.toString());
			try {
				Long queProcessID = jd.incr(RedisKeys.sProdThreadProcessID);
				String tempCommand = "nsrclone -vvv -F -C 1 -b \"" + targetClone + "\" -S " + ss.getElement();
				tx.hset(RedisKeys.hashProdThreadQueuedCommand,String.valueOf(queProcessID),cloneName + "|" + tempCommand);
				tx.hset(RedisKeys.hashProdThreadQueuedProcessID, String.valueOf(queProcessID), getScore(jd,RedisKeys.hashProdCloneSSID,ss.getElement()));
				tasks.add(Executors.callable(new WorkerThreadV2(tempCommand,String.valueOf(queProcessID))));
				tx.zrem(cloneKey, ss.getElement());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			//tx.exec();
		}
		tx.exec();


		// group by 10 if under 50GB in size

		boolean loopRun = true;


		while (loopRun) {
			s = jd.zrevrangeWithScores(cloneKey, 0 , 9);
			if (s.size() == 0) {
				loopRun = false;
				break;
			}
			long cloneJobSize = 0;
			tx = jdpl.multi();
			List<String> ssidList = new ArrayList<String>();
			for (Tuple ss : s) {
				double d = ss.getScore();
				BigDecimal big = new BigDecimal(d);
				cloneJobSize = cloneJobSize + (long) d;
				System.out.println(ss.getElement() + " " +  big.toString());
				tx.zrem(cloneKey, ss.getElement());
				ssidList.add(ss.getElement());
			}
			tx.exec();
			String ssidStr = "";
			for (String ssid : ssidList) {
				ssidStr = ssidStr + " " + ssid;
			}
			ssidStr =  ssidStr.trim();
			tx = jdpl.multi();
			try {
				Long queProcessID = jd.incr(RedisKeys.sProdThreadProcessID);
				String tempCommand = "nsrclone -vvv -F -C 1 -b \"" + targetClone + "\" -S " + ssidStr;
				tx.hset(RedisKeys.hashProdThreadQueuedCommand,String.valueOf(queProcessID),cloneName + "|" + tempCommand);
				tx.hset(RedisKeys.hashProdThreadQueuedProcessID, String.valueOf(queProcessID), String.valueOf(cloneJobSize));
				tasks.add(Executors.callable(new WorkerThreadV2(tempCommand,String.valueOf(queProcessID))));

			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			tx.exec();
			//jd.close();
			//jdpl.close();

		}

		try {
			executor.invokeAll(tasks);

		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();

		}
		executor.shutdown();

		/**


		String pushCoStatus = "Program Start : "+ programStart + "\n";
		pushCoStatus = "Program End : "+ programEnd + "\n";

		for (Tuple ss : s) {
			pushCoStatus = pushCoStatus + ss.getElement() + "\n";
		}


		sl.setLogandWrite("t:log:message", "info", "low", "cloneCatchUp", "cloneCatchUp", "cloneCatchUp-log", "Clone Complete\n" + pushCoStatus);

		programEnd = (Misc.getTimestamp());
		//new PushCo("BackupInfo","Clone Complete : " + pushCoStatus,true);
		 * 
		 * 
		 */

		//GetRedisServer grs = new GetRedisServer();
		//String redisHost = grs.getRedisHost();
		//jd = new Jedis(RedisVars.redisHost);
		//try {
		//	grs.checkForRedis(jd);
		//} catch (IOException e) {
		//	// TODO Auto-generated catch block
		//	e.printStackTrace();
		//	System.exit(1);
		//}

		jd.lrem(RedisKeys.listProdCloneRunning, 1, tempkey);
		//System.out.println("Trying to delete " + processIDStr + " from " + RedisKeys.hashProdCloneRunningName);
		jd.hdel(RedisKeys.hashProdCloneRunningName, processIDStr);
		jd.rpush(RedisKeys.lProdCloneCompleted, processIDStr + "|" + cloneName + "|" + programStart + "|" + programEnd + "|cloneCatchUp|" + targetClone);
		jd.del(cloneKey);
		System.out.println("Finished all threads");
		jd.close();
		jdpl.close();
	}

	public Long getScoreLong(Jedis jd,String key, String score) {
		return Long.parseLong(jd.hget(key, score));
	}

	public static String getScore(Jedis jd,String hash, String key) {
		return jd.hget(hash, key);
	}
}

